package concurrent.c_012_ThreadPool;

import lombok.AllArgsConstructor;

import java.io.IOException;
import java.time.Duration;
import java.time.Instant;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Random;
import java.util.concurrent.*;
import java.util.stream.LongStream;

/**
 * http://ifeve.com/forkjoin
 * https://docs.oracle.com/javase/tutorial/essential/concurrency/forkjoin.html
 * https://blog.csdn.net/f641385712/article/details/83749798
 * ForkJoinPool 不是为了替代 ExecutorService，而是它的补充，在某些应用场景下性能比 ExecutorService 更好。
 * ForkJoinPool 主要用于实现“分而治之”的算法，特别是分治之后递归调用的函数，例如 quick sort 等。
 * ForkJoinPool 最适合的是计算密集型的任务，如果存在 I/O，线程间同步，sleep() 等会造成线程长时间阻塞的情况时，最好配合使用 ManagedBlocker。
 */
public class T10_ForkJoinPool {
    static int[] nums = new int[1000000];
    static final int MAX_NUM = 50000;
    static Random r = new Random();

    static {
        for (int i = 0; i < nums.length; i++) {
            nums[i] = r.nextInt(100);
        }
        System.out.println("stream api --- " + Arrays.stream(nums).sum());
    }

    @AllArgsConstructor
    static class AddTask extends RecursiveAction {
        int start, end;

        @Override
        protected void compute() {
            if (end - start <= MAX_NUM) {
                long sum = 0L;
                for (int i = start; i < end; i++) {
                    sum += nums[i];
                }
                System.out.println("from:" + start + " to:" + end + " = " + sum);
            } else {
                int middle = start + (end - start) / 2;
                AddTask subTask1 = new AddTask(start, middle);
                AddTask subTask2 = new AddTask(middle, end);
                subTask1.fork();
                subTask2.fork();
            }
        }
    }

    @AllArgsConstructor
    static class AddTaskRet extends RecursiveTask<Long> {
        private static final long serialVersionUID = 1L;
        int start, end;

        @Override
        protected Long compute() {
            if (end - start <= MAX_NUM) {
                long sum = 0L;
                for (int i = start; i < end; i++) sum += nums[i];
                return sum;
            }

            int middle = start + (end - start) / 2;

            AddTaskRet subTask1 = new AddTaskRet(start, middle);
            AddTaskRet subTask2 = new AddTaskRet(middle, end);
            subTask1.fork();
            subTask2.fork();

            return subTask1.join() + subTask2.join();
        }
    }

    public static void main(String[] args) throws IOException {
        ForkJoinPool fjp = new ForkJoinPool();
        AddTask task = new AddTask(0, nums.length);
        // AddTaskRet task = new AddTaskRet(0, nums.length);
        fjp.execute(task);
        System.out.println(task.join());

        System.out.println("=========================================================================================");

        long[] numbers = LongStream.rangeClosed(1, 10000000).toArray();
        Instant start = Instant.now();
        // Calculator calculator = new ForLoopCalculator();
        // Calculator calculator = new ExecutorServiceCalculator();
        Calculator calculator = new ForkJoinCalculator();
        // 采用并行流，JDK8 以后的推荐做法，并行流底层还是 Fork/Join 框架，只是任务拆分优化得很好
        // long result = LongStream.rangeClosed(0, 10000000L).parallel().reduce(0, Long::sum);
        long result = calculator.sumUp(numbers);
        Instant end = Instant.now();
        System.out.println("耗时：" + Duration.between(start, end).toMillis() + "ms");
        System.out.println("结果为：" + result);
    }

    // ==============================================================================================================

    // 计算 1 至 10000000 的正整数之和
    interface Calculator { // 面向接口编程，把计算的方法定义成接口，不同的方案编码不同的实现
        long sumUp(long[] numbers); // 把传进来的所有 numbers 做求和处理
    }

    static class ForLoopCalculator implements Calculator { // 通过普通的 for 循环实现总和的相加
        @Override
        public long sumUp(long[] numbers) {
            long total = 0;
            for (long i : numbers) {
                total += i;
            }
            return total;
        }
    }

    static class ExecutorServiceCalculator implements Calculator { // 使用 ExecutorService 实现多线程的求和
        final int parallism = Runtime.getRuntime().availableProcessors(); // 默认用 CPU 核心数
        final ExecutorService pool = Executors.newFixedThreadPool(parallism);

        @Override
        public long sumUp(long[] numbers) {
            System.out.println("ExecutorService 线程池大小：" + parallism);
            List<Future<Long>> results = new ArrayList<>();

            // 把任务分解为 n 份，交给 n 个线程处理，4 核心，就等分成 4 份，然后把每一份都扔个一个 SumTask 线程进行处理
            int part = numbers.length / parallism;
            for (int i = 0; i < parallism; i++) {
                int from = i * part; // 开始位置
                int to = (i == parallism - 1) ? numbers.length - 1 : (i + 1) * part - 1; // 结束位置
                results.add(pool.submit(new SumTask(numbers, from, to))); // 扔给线程池计算
            }

            // 把每个线程的结果相加，得到最终结果 get() 方法是阻塞的，可以采用 CompletableFuture 来优化，JDK1.8 特性
            long total = 0L;
            for (Future<Long> f : results) {
                try {
                    total += f.get();
                } catch (Exception ignore) {
                }
            }
            return total;
        }

        @AllArgsConstructor
        private static class SumTask implements Callable<Long> { // 处理计算任务的线程
            final long[] numbers;
            final int from;
            final int to;

            @Override
            public Long call() {
                long total = 0;
                for (int i = from; i <= to; i++) {
                    total += numbers[i];
                }
                return total;
            }
        }
    }

    // Fork/Join 并行流等当计算的数字非常大的时候，优势才能体现出来。也就是说，如果计算比较小，或者不是 CPU 密集型的任务，不太建议使用并行处理
    static class ForkJoinCalculator implements Calculator {
        final ForkJoinPool pool = new ForkJoinPool();
        // private final ForkJoinPool pool = ForkJoinPool.commonPool(); // 可以使用公用的线程池

        @Override
        public long sumUp(long[] numbers) {
            System.out.println("ForkJoinPool 线程池大小：" + pool.getPoolSize());
            Long result = pool.invoke(new SumTask(numbers, 0, numbers.length - 1));
            pool.shutdown();
            return result;
        }

        @AllArgsConstructor
        static class SumTask extends RecursiveTask<Long> { // 执行任务 RecursiveTask：有返回值  RecursiveAction：无返回值
            final long[] numbers;
            final int from;
            final int to;

            // 没有显式地“把任务分配给线程”，只是分解了任务，而把具体的任务到线程的映射交给了 ForkJoinPool 来完成
            @Override
            protected Long compute() { // 此方法为 ForkJoin 的核心方法：对任务进行拆分，拆分的好坏决定了效率的高低
                if (to - from < 6) { // 当需要计算的数字个数小于 6 时，直接采用 for loop 方式计算结果
                    long total = 0;
                    for (int i = from; i <= to; i++) {
                        total += numbers[i];
                    }
                    return total;
                } else { // 否则，把任务一分为二，递归拆分(注意此处有递归)到底拆分成多少分需要根据具体情况而定
                    int middle = (from + to) / 2;
                    SumTask taskLeft = new SumTask(numbers, from, middle);
                    SumTask taskRight = new SumTask(numbers, middle + 1, to);
                    taskLeft.fork();
                    taskRight.fork();
                    return taskLeft.join() + taskRight.join();
                }
            }
        }
    }
}
